package com.taobao.analyse.utils;

import java.io.ByteArrayInputStream;
import java.io.DataInput;
import java.io.DataInputStream;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.io.DataInputBuffer;
import org.apache.hadoop.mapred.Reporter;

import com.taobao.cmp.citadel.indicator.DataType;
import com.taobao.cmp.citadel.indicator.Indicator;
import com.taobao.cmp.citadel.indicator.IndicatorConf;
import com.taobao.cmp.citadel.indicator.IndicatorLoader;
import com.taobao.cmp.citadel.message.MessageType;
import com.taobao.cmp.citadel.message.PropConstants;

/**
 * @author xuanwei
 * @version 2011-10-20
 * @Mail-To xuanwei@taobao.com
 */
public class LCDataParser {
	private static final IndicatorConf		INDI_CONF		= IndicatorLoader.loadIndicators();
	private static final Map<Byte, String>	messageTypes	= new HashMap<Byte, String>();

	static {
		messageTypes.put((byte) 0, MessageType.IC_CC);
		messageTypes.put((byte) 1, MessageType.IC_CHANGE_QUANTITY);
		messageTypes.put((byte) 2, MessageType.IC_FEATURE_CHANAGE);
		messageTypes.put((byte) 3, MessageType.IC_PUBLISH_ITEM);
		messageTypes.put((byte) 4, MessageType.IC_SELLER_DEL);
		messageTypes.put((byte) 5, MessageType.IC_SELLER_DOWNSHELF);
		messageTypes.put((byte) 6, MessageType.IC_SELLER_EDIT);
		messageTypes.put((byte) 7, MessageType.IC_SELLER_UPSHELF);
		messageTypes.put((byte) 8, MessageType.IC_XIAOER_DEL);
		messageTypes.put((byte) 9, MessageType.IC_XIAOER_DOWNSHELF);
		messageTypes.put((byte) 10, MessageType.IC_XIAOER_EDIT);
		messageTypes.put((byte) 11, MessageType.IC_ZERO_QUANTITY);

		messageTypes.put((byte) 12, MessageType.LOOM_B_DEL_COUNT);
		messageTypes.put((byte) 13, MessageType.LOOM_B_DEL_UNCOUNT);
		messageTypes.put((byte) 14, MessageType.LOOM_B_DOWNSHELF_COUNT);
		messageTypes.put((byte) 15, MessageType.LOOM_B_DOWNSHELF_UNCOUNT);
		messageTypes.put((byte) 16, MessageType.LOOM_CC);
		messageTypes.put((byte) 17, MessageType.LOOM_DEL_COUNT);
		messageTypes.put((byte) 18, MessageType.LOOM_DEL_UNCOUNT);
		messageTypes.put((byte) 19, MessageType.LOOM_DOWNSHELF_COUNT);
		messageTypes.put((byte) 20, MessageType.LOOM_DOWNSHELF_UNCOUNT);
		messageTypes.put((byte) 21, MessageType.LOOM_REDUCE_WEIGHT);
		messageTypes.put((byte) 22, MessageType.LOOM_SHIELD);
		messageTypes.put((byte) 23, MessageType.LOOM_CONFIRM);
	}

	public static Map<String, Object> read(byte[] bytes, Reporter reporter) throws IOException {
		DataInputStream in = new DataInputStream(new ByteArrayInputStream(bytes));
		Map<String, Object> map = new HashMap<String, Object>();

		map.put("sellerId", in.readLong());
		map.put("itemId", in.readLong());
		map.put("time", in.readLong());

		while (in.available() > 0) {
			short id = in.readShort();
			Indicator indi = INDI_CONF.getIndicator(id);
			Object obj = readObject(indi, in, map, reporter);
			map.put(indi.getName(), obj);
		}

		return map;
	}

	public static String toString(final byte[] bytes) {
		return toString(bytes, "0x", 0, bytes.length);
	}

	public static String toString(final byte[] bytes, String prefix, int off, int len) {
		StringBuilder sb = new StringBuilder();
		sb.append("[ ");
		for (int i = off; i < off + len; ++i) {
			sb.append(prefix);
			sb.append(Integer.toHexString((bytes[i] & 0xff) >> 4));
			sb.append(Integer.toHexString((bytes[i] & 0xff) % 16));
			sb.append(" ");
		}
		sb.append(']');
		return sb.toString();
	}

	public static Map<String, Object> read(DataInputBuffer in, Reporter reporter) throws IOException {
		Map<String, Object> map = new HashMap<String, Object>();

		map.put("sellerId", in.readLong());
		map.put("itemId", in.readLong());
		map.put("time", in.readLong());
		while (in.available() > 0) {
			short id = in.readShort();
			Indicator indi = INDI_CONF.getIndicator(id);
			Object obj = readObject(indi, in, map, reporter);
			map.put(indi.getName(), obj);
		}
		return map;
	}

	public static Object readObject(Indicator indicator, DataInput in, Map<String, Object> map, Reporter reporter)
			throws IOException {
		if (indicator == null) {
			reporter.incrCounter("DupPunish", "NullIndicator", 1);
			return null;
		}
		if (PropConstants.MESSAGE_TYPE.equals(indicator.getName())) {
			byte b = in.readByte();
			return toMessageType(b);
		}
		DataType type = indicator.getType();
		if (type == DataType.BYTE) {
			return in.readByte();
		} else if (type == DataType.INT) {
			return in.readInt();
		} else if (type == DataType.FLOAT) {
			return in.readFloat();
		} else if (type == DataType.SHORT) {
			return in.readShort();
		} else if (type == DataType.DOUBLE) {
			return in.readDouble();
		} else if (type == DataType.LONG) {
			return in.readLong();
		} else if (type == DataType.STRING) {
			int length = in.readInt();
			byte[] bytes = new byte[length];
			in.readFully(bytes);
			return new String(bytes, "GBK");
		} else if (type == DataType.DATE) {
			return new Date(in.readLong());
		} else if (type == DataType.BYTES) {
			int length = in.readInt();
			byte[] data = new byte[length];
			in.readFully(data);
			return data;
		}
		return null;
	}

	private static String toMessageType(byte b) {
		if (messageTypes.containsKey(b)) {
			return messageTypes.get(b);
		}
		return "default";
	}
}
